package io.quarkus.smallrye.reactivemessaging.kafka.deployment;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.groups.Tuple.tuple;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletionStage;
import java.util.function.Function;

import jakarta.enterprise.inject.Instance;
import jakarta.inject.Inject;

import org.apache.avro.generic.GenericRecord;
import org.apache.avro.specific.AvroGenerated;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serializer;
import org.assertj.core.api.Assert;
import org.assertj.core.groups.Tuple;
import org.eclipse.microprofile.config.Config;
import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Emitter;
import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.messaging.Message;
import org.eclipse.microprofile.reactive.messaging.Outgoing;
import org.eclipse.microprofile.reactive.streams.operators.ProcessorBuilder;
import org.eclipse.microprofile.reactive.streams.operators.PublisherBuilder;
import org.eclipse.microprofile.reactive.streams.operators.SubscriberBuilder;
import org.jboss.jandex.IndexView;
import org.jboss.jandex.Indexer;
import org.junit.jupiter.api.Test;
import org.reactivestreams.Processor;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;

import io.quarkus.arc.InjectableInstance;
import io.quarkus.commons.classloading.ClassLoaderHelper;
import io.quarkus.deployment.builditem.GeneratedClassBuildItem;
import io.quarkus.deployment.builditem.RunTimeConfigurationDefaultBuildItem;
import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem;
import io.quarkus.kafka.client.serialization.JsonbSerializer;
import io.quarkus.kafka.client.serialization.ObjectMapperDeserializer;
import io.quarkus.runtime.configuration.QuarkusConfigFactory;
import io.quarkus.smallrye.reactivemessaging.deployment.items.ConnectorManagedChannelBuildItem;
import io.smallrye.config.SmallRyeConfigBuilder;
import io.smallrye.config.common.MapBackedConfigSource;
import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.operators.multi.split.MultiSplitter;
import io.smallrye.reactive.messaging.GenericPayload;
import io.smallrye.reactive.messaging.MutinyEmitter;
import io.smallrye.reactive.messaging.Targeted;
import io.smallrye.reactive.messaging.TargetedMessages;
import io.smallrye.reactive.messaging.kafka.KafkaRecord;
import io.smallrye.reactive.messaging.kafka.KafkaRecordBatch;
import io.smallrye.reactive.messaging.kafka.Record;
import io.smallrye.reactive.messaging.kafka.reply.KafkaRequestReply;
import io.smallrye.reactive.messaging.kafka.transactions.KafkaTransactions;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;

public class DefaultSerdeConfigTest {
    private static void doTest(Tuple[] expectations, Class<?>... classesToIndex) {
        doTest(null, expectations, classesToIndex);
    }

    private static void doTest(Config customConfig, Tuple[] expectations, Class<?>... classesToIndex) {
        doTest(customConfig, expectations, Collections.emptyList(), Collections.emptyList(), classesToIndex);
    }

    @SuppressWarnings({ "unchecked", "rawtypes" })
    private static void doTest(Config customConfig, Tuple[] expectations,
            List<Function<String, Assert>> generatedNames,
            List<Function<String, Assert>> reflectiveNames, Class<?>... classesToIndex) {
        List<RunTimeConfigurationDefaultBuildItem> configs = new ArrayList<>();
        List<GeneratedClassBuildItem> generated = new ArrayList<>();
        List<ReflectiveClassBuildItem> reflective = new ArrayList<>();

        List<Class<?>> classes = new ArrayList<>(Arrays.asList(classesToIndex));
        classes.add(Incoming.class);
        classes.add(Outgoing.class);
        classes.add(Serializer.class);
        classes.add(Deserializer.class);
        classes.add(io.quarkus.kafka.client.serialization.ObjectMapperDeserializer.class);
        classes.add(io.quarkus.kafka.client.serialization.ObjectMapperSerializer.class);
        classes.add(io.quarkus.kafka.client.serialization.JsonbSerializer.class);
        classes.add(io.quarkus.kafka.client.serialization.JsonbDeserializer.class);
        DefaultSerdeDiscoveryState discovery = new DefaultSerdeDiscoveryState(index(classes)) {
            @Override
            Config getConfig() {
                return customConfig != null ? customConfig : super.getConfig();
            }

            @Override
            boolean isKafkaConnector(List<ConnectorManagedChannelBuildItem> list, boolean incoming, String channelName) {
                return true;
            }
        };
        try {
            new SmallRyeReactiveMessagingKafkaProcessor().discoverDefaultSerdeConfig(discovery, Collections.emptyList(),
                    configs::add,
                    (generatedNames == null) ? null : generated::add,
                    (reflectiveNames == null) ? null : reflective::add);

            assertThat(configs)
                    .extracting(RunTimeConfigurationDefaultBuildItem::getKey, RunTimeConfigurationDefaultBuildItem::getValue)
                    .hasSize(expectations.length)
                    .allSatisfy(tuple -> {
                        Object[] e = tuple.toArray();
                        String key = (String) e[0];
                        String value = (String) e[1];
                        assertThat(Arrays.stream(expectations).filter(t -> key.equals(t.toArray()[0])))
                                .hasSize(1)
                                .satisfiesOnlyOnce(t -> {
                                    Object o = t.toArray()[1];
                                    if (o instanceof String) {
                                        assertThat(value).isEqualTo((String) o);
                                    } else {
                                        ((Function<String, Assert>) o).apply(value);
                                    }
                                });
                    });

            assertThat(generated)
                    .extracting(GeneratedClassBuildItem::internalName)
                    .allSatisfy(s -> assertThat(generatedNames).satisfiesOnlyOnce(c -> c.apply(s)));

            assertThat(reflective)
                    .flatExtracting(ReflectiveClassBuildItem::getClassNames)
                    .extracting(n -> n.replace('/', '.'))
                    .allSatisfy(s -> assertThat(reflectiveNames).satisfiesOnlyOnce(c -> c.apply(s)));
        } finally {
            // must not leak the lazily-initialized Config instance associated to the system classloader
            if (customConfig == null) {
                QuarkusConfigFactory.setConfig(null);
            }
        }
    }

    Function<String, Assert> assertMatches(String regex) {
        return s -> assertThat(s).matches(regex);
    }

    Function<String, Assert> assertStartsWith(String starts) {
        return s -> assertThat(s).startsWith(starts);
    }

    private static IndexView index(List<Class<?>> classes) {
        Indexer indexer = new Indexer();
        for (Class<?> clazz : classes) {
            final String resourceName = ClassLoaderHelper.fromClassNameToResourceName(clazz.getName());
            try {
                try (InputStream stream = DefaultSerdeConfigTest.class.getClassLoader()
                        .getResourceAsStream(resourceName)) {
                    indexer.index(stream);
                }
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
        return indexer.complete();
    }

    // ---

    @Test
    public void stringInLongOut() {
        // @formatter:off
        Tuple[] expectations = {
                tuple("mp.messaging.outgoing.channel1.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.outgoing.channel2.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.outgoing.channel3.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.outgoing.channel4.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.outgoing.channel5.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.outgoing.channel6.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.outgoing.channel7.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.outgoing.channel8.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.outgoing.channel9.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.outgoing.channel10.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.outgoing.channel11.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.outgoing.channel12.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"),

                tuple("mp.messaging.incoming.channel13.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.incoming.channel14.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.incoming.channel15.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.incoming.channel16.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.incoming.channel17.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.incoming.channel18.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.incoming.channel19.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.incoming.channel20.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.incoming.channel21.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),

                tuple("mp.messaging.incoming.channel22.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel23.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel24.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel25.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel26.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel27.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel28.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel29.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel30.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel31.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel32.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel33.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel34.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel35.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel36.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel37.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel38.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel39.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel40.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel41.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel42.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel43.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel44.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel45.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel46.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel47.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel48.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel49.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel50.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel51.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel52.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel53.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),

                tuple("mp.messaging.incoming.channel54.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel55.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel56.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel57.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel58.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel59.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel60.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel61.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel62.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel63.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel64.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel65.value.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
        };
        // @formatter:on

        doTest(expectations, StringInLongOut.class);
    }

    private static class StringInLongOut {
        // @Outgoing

        @Outgoing("channel1")
        Publisher<Message<Long>> method1() {
            return null;
        }

        @Outgoing("channel2")
        Publisher<Long> method2() {
            return null;
        }

        @Outgoing("channel3")
        PublisherBuilder<Message<Long>> method3() {
            return null;
        }

        @Outgoing("channel4")
        PublisherBuilder<Long> method4() {
            return null;
        }

        @Outgoing("channel5")
        Multi<Message<Long>> method5() {
            return null;
        }

        @Outgoing("channel6")
        Multi<Long> method6() {
            return null;
        }

        @Outgoing("channel7")
        Message<Long> method7() {
            return null;
        }

        @Outgoing("channel8")
        Long method8() {
            return null;
        }

        @Outgoing("channel9")
        CompletionStage<Message<Long>> method9() {
            return null;
        }

        @Outgoing("channel10")
        CompletionStage<Long> method10() {
            return null;
        }

        @Outgoing("channel11")
        Uni<Message<Long>> method11() {
            return null;
        }

        @Outgoing("channel12")
        Uni<Long> method12() {
            return null;
        }

        // @Incoming

        @Incoming("channel13")
        Subscriber<Message<String>> method13() {
            return null;
        }

        @Incoming("channel14")
        Subscriber<String> method14() {
            return null;
        }

        @Incoming("channel15")
        SubscriberBuilder<Message<String>, Void> method15() {
            return null;
        }

        @Incoming("channel16")
        SubscriberBuilder<String, Void> method16() {
            return null;
        }

        @Incoming("channel17")
        void method17(String msg) {
        }

        @Incoming("channel18")
        CompletionStage<?> method18(Message<String> msg) {
            return null;
        }

        @Incoming("channel19")
        CompletionStage<?> method19(String payload) {
            return null;
        }

        @Incoming("channel20")
        Uni<?> method20(Message<String> msg) {
            return null;
        }

        @Incoming("channel21")
        Uni<?> method21(String payload) {
            return null;
        }

        // @Incoming @Outgoing

        @Incoming("channel22")
        @Outgoing("channel23")
        Processor<Message<String>, Message<Long>> method22() {
            return null;
        }

        @Incoming("channel24")
        @Outgoing("channel25")
        Processor<String, Long> method23() {
            return null;
        }

        @Incoming("channel26")
        @Outgoing("channel27")
        ProcessorBuilder<Message<String>, Message<Long>> method24() {
            return null;
        }

        @Incoming("channel28")
        @Outgoing("channel29")
        ProcessorBuilder<String, Long> method25() {
            return null;
        }

        @Incoming("channel30")
        @Outgoing("channel31")
        Publisher<Message<Long>> method26(Message<String> msg) {
            return null;
        }

        @Incoming("channel32")
        @Outgoing("channel33")
        Publisher<Long> method27(String payload) {
            return null;
        }

        @Incoming("channel34")
        @Outgoing("channel35")
        PublisherBuilder<Message<Long>> method28(Message<String> msg) {
            return null;
        }

        @Incoming("channel36")
        @Outgoing("channel37")
        PublisherBuilder<Long> method29(String payload) {
            return null;
        }

        @Incoming("channel38")
        @Outgoing("channel39")
        Multi<Message<Long>> method30(Message<String> msg) {
            return null;
        }

        @Incoming("channel40")
        @Outgoing("channel41")
        Multi<Long> method31(String payload) {
            return null;
        }

        @Incoming("channel42")
        @Outgoing("channel43")
        Message<Long> method32(Message<String> msg) {
            return null;
        }

        @Incoming("channel44")
        @Outgoing("channel45")
        Long method33(String payload) {
            return null;
        }

        @Incoming("channel46")
        @Outgoing("channel47")
        CompletionStage<Message<Long>> method34(Message<String> msg) {
            return null;
        }

        @Incoming("channel48")
        @Outgoing("channel49")
        CompletionStage<Long> method35(String payload) {
            return null;
        }

        @Incoming("channel50")
        @Outgoing("channel51")
        Uni<Message<Long>> method36(Message<String> msg) {
            return null;
        }

        @Incoming("channel52")
        @Outgoing("channel53")
        Uni<Long> method37(String payload) {
            return null;
        }

        // @Incoming @Outgoing stream manipulation

        @Incoming("channel54")
        @Outgoing("channel55")
        Publisher<Message<Long>> method38(Publisher<Message<String>> msg) {
            return null;
        }

        @Incoming("channel56")
        @Outgoing("channel57")
        Publisher<Long> method39(Publisher<String> payload) {
            return null;
        }

        @Incoming("channel58")
        @Outgoing("channel59")
        PublisherBuilder<Message<Long>> method40(PublisherBuilder<Message<String>> msg) {
            return null;
        }

        @Incoming("channel60")
        @Outgoing("channel61")
        PublisherBuilder<Long> method41(PublisherBuilder<String> payload) {
            return null;
        }

        @Incoming("channel62")
        @Outgoing("channel63")
        Multi<Message<Long>> method42(Multi<Message<String>> msg) {
            return null;
        }

        @Incoming("channel64")
        @Outgoing("channel65")
        Multi<Long> method43(Multi<String> payload) {
            return null;
        }
    }

    // ---

    @Test
    public void byteArrayInAvroDtoOut() {
        // @formatter:off
        Tuple[] expectations = {
                tuple("mp.messaging.outgoing.channel1.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel2.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel3.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel4.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel5.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel6.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel7.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel8.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel9.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel10.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel11.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel12.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),

                tuple("mp.messaging.incoming.channel13.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.incoming.channel14.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.incoming.channel15.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.incoming.channel16.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.incoming.channel17.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.incoming.channel18.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.incoming.channel19.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.incoming.channel20.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.incoming.channel21.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),

                tuple("mp.messaging.incoming.channel22.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel23.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel24.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel25.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel26.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel27.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel28.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel29.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel30.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel31.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel32.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel33.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel34.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel35.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel36.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel37.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel38.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel39.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel40.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel41.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel42.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel43.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel44.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel45.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel46.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel47.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel48.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel49.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel50.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel51.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel52.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel53.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),

                tuple("mp.messaging.incoming.channel54.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel55.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel56.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel57.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel58.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel59.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel60.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel61.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel62.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel63.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel64.value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer"),
                tuple("mp.messaging.outgoing.channel65.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
        };
        // @formatter:on

        doTest(expectations, AvroDto.class, ByteArrayInAvroDtoOut.class);
    }

    // simulating an Avro-generated class, the autodetection code only looks for this annotation
    @AvroGenerated
    private static class AvroDto {
    }

    private static class ByteArrayInAvroDtoOut {
        // @Outgoing

        @Outgoing("channel1")
        Publisher<Message<AvroDto>> method1() {
            return null;
        }

        @Outgoing("channel2")
        Publisher<AvroDto> method2() {
            return null;
        }

        @Outgoing("channel3")
        PublisherBuilder<Message<AvroDto>> method3() {
            return null;
        }

        @Outgoing("channel4")
        PublisherBuilder<AvroDto> method4() {
            return null;
        }

        @Outgoing("channel5")
        Multi<Message<AvroDto>> method5() {
            return null;
        }

        @Outgoing("channel6")
        Multi<AvroDto> method6() {
            return null;
        }

        @Outgoing("channel7")
        Message<AvroDto> method7() {
            return null;
        }

        @Outgoing("channel8")
        AvroDto method8() {
            return null;
        }

        @Outgoing("channel9")
        CompletionStage<Message<AvroDto>> method9() {
            return null;
        }

        @Outgoing("channel10")
        CompletionStage<AvroDto> method10() {
            return null;
        }

        @Outgoing("channel11")
        Uni<Message<AvroDto>> method11() {
            return null;
        }

        @Outgoing("channel12")
        Uni<AvroDto> method12() {
            return null;
        }

        // @Incoming

        @Incoming("channel13")
        Subscriber<Message<byte[]>> method13() {
            return null;
        }

        @Incoming("channel14")
        Subscriber<byte[]> method14() {
            return null;
        }

        @Incoming("channel15")
        SubscriberBuilder<Message<byte[]>, Void> method15() {
            return null;
        }

        @Incoming("channel16")
        SubscriberBuilder<byte[], Void> method16() {
            return null;
        }

        @Incoming("channel17")
        void method17(byte[] msg) {
        }

        @Incoming("channel18")
        CompletionStage<?> method18(Message<byte[]> msg) {
            return null;
        }

        @Incoming("channel19")
        CompletionStage<?> method19(byte[] payload) {
            return null;
        }

        @Incoming("channel20")
        Uni<?> method20(Message<byte[]> msg) {
            return null;
        }

        @Incoming("channel21")
        Uni<?> method21(byte[] payload) {
            return null;
        }

        // @Incoming @Outgoing

        @Incoming("channel22")
        @Outgoing("channel23")
        Processor<Message<byte[]>, Message<AvroDto>> method22() {
            return null;
        }

        @Incoming("channel24")
        @Outgoing("channel25")
        Processor<byte[], AvroDto> method23() {
            return null;
        }

        @Incoming("channel26")
        @Outgoing("channel27")
        ProcessorBuilder<Message<byte[]>, Message<AvroDto>> method24() {
            return null;
        }

        @Incoming("channel28")
        @Outgoing("channel29")
        ProcessorBuilder<byte[], AvroDto> method25() {
            return null;
        }

        @Incoming("channel30")
        @Outgoing("channel31")
        Publisher<Message<AvroDto>> method26(Message<byte[]> msg) {
            return null;
        }

        @Incoming("channel32")
        @Outgoing("channel33")
        Publisher<AvroDto> method27(byte[] payload) {
            return null;
        }

        @Incoming("channel34")
        @Outgoing("channel35")
        PublisherBuilder<Message<AvroDto>> method28(Message<byte[]> msg) {
            return null;
        }

        @Incoming("channel36")
        @Outgoing("channel37")
        PublisherBuilder<AvroDto> method29(byte[] payload) {
            return null;
        }

        @Incoming("channel38")
        @Outgoing("channel39")
        Multi<Message<AvroDto>> method30(Message<byte[]> msg) {
            return null;
        }

        @Incoming("channel40")
        @Outgoing("channel41")
        Multi<AvroDto> method31(byte[] payload) {
            return null;
        }

        @Incoming("channel42")
        @Outgoing("channel43")
        Message<AvroDto> method32(Message<byte[]> msg) {
            return null;
        }

        @Incoming("channel44")
        @Outgoing("channel45")
        AvroDto method33(byte[] payload) {
            return null;
        }

        @Incoming("channel46")
        @Outgoing("channel47")
        CompletionStage<Message<AvroDto>> method34(Message<byte[]> msg) {
            return null;
        }

        @Incoming("channel48")
        @Outgoing("channel49")
        CompletionStage<AvroDto> method35(byte[] payload) {
            return null;
        }

        @Incoming("channel50")
        @Outgoing("channel51")
        Uni<Message<AvroDto>> method36(Message<byte[]> msg) {
            return null;
        }

        @Incoming("channel52")
        @Outgoing("channel53")
        Uni<AvroDto> method37(byte[] payload) {
            return null;
        }

        // @Incoming @Outgoing stream manipulation

        @Incoming("channel54")
        @Outgoing("channel55")
        Publisher<Message<AvroDto>> method38(Publisher<Message<byte[]>> msg) {
            return null;
        }

        @Incoming("channel56")
        @Outgoing("channel57")
        Publisher<AvroDto> method39(Publisher<byte[]> payload) {
            return null;
        }

        @Incoming("channel58")
        @Outgoing("channel59")
        PublisherBuilder<Message<AvroDto>> method40(PublisherBuilder<Message<byte[]>> msg) {
            return null;
        }

        @Incoming("channel60")
        @Outgoing("channel61")
        PublisherBuilder<AvroDto> method41(PublisherBuilder<byte[]> payload) {
            return null;
        }

        @Incoming("channel62")
        @Outgoing("channel63")
        Multi<Message<AvroDto>> method42(Multi<Message<byte[]>> msg) {
            return null;
        }

        @Incoming("channel64")
        @Outgoing("channel65")
        Multi<AvroDto> method43(Multi<byte[]> payload) {
            return null;
        }
    }

    // ---

    @Test
    public void avroDtoInGenericRecordOut() {
        // @formatter:off
        Tuple[] expectations = {
                tuple("mp.messaging.outgoing.channel1.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel2.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel3.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel4.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel5.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel6.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel7.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel8.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel9.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel10.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel11.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel12.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),

                tuple("mp.messaging.incoming.channel13.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel13.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.incoming.channel14.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel14.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.incoming.channel15.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel15.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.incoming.channel16.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel16.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.incoming.channel17.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel17.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.incoming.channel18.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel18.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.incoming.channel19.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel19.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.incoming.channel20.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel20.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.incoming.channel21.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel21.apicurio.registry.use-specific-avro-reader", "true"),

                tuple("mp.messaging.incoming.channel22.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel22.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel23.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel24.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel24.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel25.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel26.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel26.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel27.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel28.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel28.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel29.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel30.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel30.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel31.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel32.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel32.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel33.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel34.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel34.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel35.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel36.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel36.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel37.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel38.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel38.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel39.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel40.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel40.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel41.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel42.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel42.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel43.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel44.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel44.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel45.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel46.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel46.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel47.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel48.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel48.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel49.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel50.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel50.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel51.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel52.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel52.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel53.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),

                tuple("mp.messaging.incoming.channel54.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel54.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel55.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel56.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel56.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel57.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel58.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel58.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel59.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel60.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel60.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel61.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel62.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel62.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel63.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.incoming.channel64.value.deserializer", "io.apicurio.registry.serde.avro.AvroKafkaDeserializer"),
                tuple("mp.messaging.incoming.channel64.apicurio.registry.use-specific-avro-reader", "true"),
                tuple("mp.messaging.outgoing.channel65.value.serializer",   "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
        };
        // @formatter:on

        doTest(expectations, AvroDto.class, AvroDtoInGenericRecordOut.class);
    }

    private static class AvroDtoInGenericRecordOut {
        // @Outgoing

        @Outgoing("channel1")
        Publisher<Message<GenericRecord>> method1() {
            return null;
        }

        @Outgoing("channel2")
        Publisher<GenericRecord> method2() {
            return null;
        }

        @Outgoing("channel3")
        PublisherBuilder<Message<GenericRecord>> method3() {
            return null;
        }

        @Outgoing("channel4")
        PublisherBuilder<GenericRecord> method4() {
            return null;
        }

        @Outgoing("channel5")
        Multi<Message<GenericRecord>> method5() {
            return null;
        }

        @Outgoing("channel6")
        Multi<GenericRecord> method6() {
            return null;
        }

        @Outgoing("channel7")
        Message<GenericRecord> method7() {
            return null;
        }

        @Outgoing("channel8")
        GenericRecord method8() {
            return null;
        }

        @Outgoing("channel9")
        CompletionStage<Message<GenericRecord>> method9() {
            return null;
        }

        @Outgoing("channel10")
        CompletionStage<GenericRecord> method10() {
            return null;
        }

        @Outgoing("channel11")
        Uni<Message<GenericRecord>> method11() {
            return null;
        }

        @Outgoing("channel12")
        Uni<GenericRecord> method12() {
            return null;
        }

        // @Incoming

        @Incoming("channel13")
        Subscriber<Message<AvroDto>> method13() {
            return null;
        }

        @Incoming("channel14")
        Subscriber<AvroDto> method14() {
            return null;
        }

        @Incoming("channel15")
        SubscriberBuilder<Message<AvroDto>, Void> method15() {
            return null;
        }

        @Incoming("channel16")
        SubscriberBuilder<AvroDto, Void> method16() {
            return null;
        }

        @Incoming("channel17")
        void method17(AvroDto msg) {
        }

        @Incoming("channel18")
        CompletionStage<?> method18(Message<AvroDto> msg) {
            return null;
        }

        @Incoming("channel19")
        CompletionStage<?> method19(AvroDto payload) {
            return null;
        }

        @Incoming("channel20")
        Uni<?> method20(Message<AvroDto> msg) {
            return null;
        }

        @Incoming("channel21")
        Uni<?> method21(AvroDto payload) {
            return null;
        }

        // @Incoming @Outgoing

        @Incoming("channel22")
        @Outgoing("channel23")
        Processor<Message<AvroDto>, Message<GenericRecord>> method22() {
            return null;
        }

        @Incoming("channel24")
        @Outgoing("channel25")
        Processor<AvroDto, GenericRecord> method23() {
            return null;
        }

        @Incoming("channel26")
        @Outgoing("channel27")
        ProcessorBuilder<Message<AvroDto>, Message<GenericRecord>> method24() {
            return null;
        }

        @Incoming("channel28")
        @Outgoing("channel29")
        ProcessorBuilder<AvroDto, GenericRecord> method25() {
            return null;
        }

        @Incoming("channel30")
        @Outgoing("channel31")
        Publisher<Message<GenericRecord>> method26(Message<AvroDto> msg) {
            return null;
        }

        @Incoming("channel32")
        @Outgoing("channel33")
        Publisher<GenericRecord> method27(AvroDto payload) {
            return null;
        }

        @Incoming("channel34")
        @Outgoing("channel35")
        PublisherBuilder<Message<GenericRecord>> method28(Message<AvroDto> msg) {
            return null;
        }

        @Incoming("channel36")
        @Outgoing("channel37")
        PublisherBuilder<GenericRecord> method29(AvroDto payload) {
            return null;
        }

        @Incoming("channel38")
        @Outgoing("channel39")
        Multi<Message<GenericRecord>> method30(Message<AvroDto> msg) {
            return null;
        }

        @Incoming("channel40")
        @Outgoing("channel41")
        Multi<GenericRecord> method31(AvroDto payload) {
            return null;
        }

        @Incoming("channel42")
        @Outgoing("channel43")
        Message<GenericRecord> method32(Message<AvroDto> msg) {
            return null;
        }

        @Incoming("channel44")
        @Outgoing("channel45")
        GenericRecord method33(AvroDto payload) {
            return null;
        }

        @Incoming("channel46")
        @Outgoing("channel47")
        CompletionStage<Message<GenericRecord>> method34(Message<AvroDto> msg) {
            return null;
        }

        @Incoming("channel48")
        @Outgoing("channel49")
        CompletionStage<GenericRecord> method35(AvroDto payload) {
            return null;
        }

        @Incoming("channel50")
        @Outgoing("channel51")
        Uni<Message<GenericRecord>> method36(Message<AvroDto> msg) {
            return null;
        }

        @Incoming("channel52")
        @Outgoing("channel53")
        Uni<GenericRecord> method37(AvroDto payload) {
            return null;
        }

        // @Incoming @Outgoing stream manipulation

        @Incoming("channel54")
        @Outgoing("channel55")
        Publisher<Message<GenericRecord>> method38(Publisher<Message<AvroDto>> msg) {
            return null;
        }

        @Incoming("channel56")
        @Outgoing("channel57")
        Publisher<GenericRecord> method39(Publisher<AvroDto> payload) {
            return null;
        }

        @Incoming("channel58")
        @Outgoing("channel59")
        PublisherBuilder<Message<GenericRecord>> method40(PublisherBuilder<Message<AvroDto>> msg) {
            return null;
        }

        @Incoming("channel60")
        @Outgoing("channel61")
        PublisherBuilder<GenericRecord> method41(PublisherBuilder<AvroDto> payload) {
            return null;
        }

        @Incoming("channel62")
        @Outgoing("channel63")
        Multi<Message<GenericRecord>> method42(Multi<Message<AvroDto>> msg) {
            return null;
        }

        @Incoming("channel64")
        @Outgoing("channel65")
        Multi<GenericRecord> method43(Multi<AvroDto> payload) {
            return null;
        }
    }

    // ---

    @Test
    public void jacksonDtoInVertxJsonObjectOut() {
        // @formatter:off
        Tuple[] expectations = {
                tuple("mp.messaging.outgoing.channel1.value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.outgoing.channel2.value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.outgoing.channel3.value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.outgoing.channel4.value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.outgoing.channel5.value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.outgoing.channel6.value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.outgoing.channel7.value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.outgoing.channel8.value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.outgoing.channel9.value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.outgoing.channel10.value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.outgoing.channel11.value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.outgoing.channel12.value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),

                tuple("mp.messaging.incoming.channel13.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.incoming.channel14.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.incoming.channel15.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.incoming.channel16.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.incoming.channel17.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.incoming.channel18.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.incoming.channel19.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.incoming.channel20.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.incoming.channel21.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),

                tuple("mp.messaging.incoming.channel22.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel23.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel24.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel25.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel26.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel27.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel28.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel29.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel30.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel31.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel32.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel33.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel34.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel35.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel36.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel37.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel38.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel39.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel40.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel41.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel42.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel43.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel44.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel45.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel46.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel47.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel48.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel49.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel50.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel51.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel52.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel53.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),

                tuple("mp.messaging.incoming.channel54.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel55.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel56.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel57.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel58.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel59.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel60.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel61.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel62.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel63.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel64.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel65.value.serializer",   "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
        };
        // @formatter:on

        doTest(expectations, JacksonDto.class, JacksonDtoDeserializer.class, JacksonDtoInVertxJsonObjectOut.class);
    }

    static class JacksonDto {
    }

    static class JacksonDtoDeserializer extends ObjectMapperDeserializer<JacksonDto> {
        public JacksonDtoDeserializer() {
            super(JacksonDto.class);
        }

        @Override
        public JacksonDto deserialize(String topic, Headers headers, byte[] data) {
            return null;
        }
    }

    private static class JacksonDtoInVertxJsonObjectOut {
        // @Outgoing

        @Outgoing("channel1")
        Publisher<Message<io.vertx.core.json.JsonObject>> method1() {
            return null;
        }

        @Outgoing("channel2")
        Publisher<io.vertx.core.json.JsonObject> method2() {
            return null;
        }

        @Outgoing("channel3")
        PublisherBuilder<Message<io.vertx.core.json.JsonObject>> method3() {
            return null;
        }

        @Outgoing("channel4")
        PublisherBuilder<io.vertx.core.json.JsonObject> method4() {
            return null;
        }

        @Outgoing("channel5")
        Multi<Message<io.vertx.core.json.JsonObject>> method5() {
            return null;
        }

        @Outgoing("channel6")
        Multi<io.vertx.core.json.JsonObject> method6() {
            return null;
        }

        @Outgoing("channel7")
        Message<io.vertx.core.json.JsonObject> method7() {
            return null;
        }

        @Outgoing("channel8")
        io.vertx.core.json.JsonObject method8() {
            return null;
        }

        @Outgoing("channel9")
        CompletionStage<Message<io.vertx.core.json.JsonObject>> method9() {
            return null;
        }

        @Outgoing("channel10")
        CompletionStage<io.vertx.core.json.JsonObject> method10() {
            return null;
        }

        @Outgoing("channel11")
        Uni<Message<io.vertx.core.json.JsonObject>> method11() {
            return null;
        }

        @Outgoing("channel12")
        Uni<io.vertx.core.json.JsonObject> method12() {
            return null;
        }

        // @Incoming

        @Incoming("channel13")
        Subscriber<Message<JacksonDto>> method13() {
            return null;
        }

        @Incoming("channel14")
        Subscriber<JacksonDto> method14() {
            return null;
        }

        @Incoming("channel15")
        SubscriberBuilder<Message<JacksonDto>, Void> method15() {
            return null;
        }

        @Incoming("channel16")
        SubscriberBuilder<JacksonDto, Void> method16() {
            return null;
        }

        @Incoming("channel17")
        void method17(JacksonDto msg) {
        }

        @Incoming("channel18")
        CompletionStage<?> method18(Message<JacksonDto> msg) {
            return null;
        }

        @Incoming("channel19")
        CompletionStage<?> method19(JacksonDto payload) {
            return null;
        }

        @Incoming("channel20")
        Uni<?> method20(Message<JacksonDto> msg) {
            return null;
        }

        @Incoming("channel21")
        Uni<?> method21(JacksonDto payload) {
            return null;
        }

        // @Incoming @Outgoing

        @Incoming("channel22")
        @Outgoing("channel23")
        Processor<Message<JacksonDto>, Message<io.vertx.core.json.JsonObject>> method22() {
            return null;
        }

        @Incoming("channel24")
        @Outgoing("channel25")
        Processor<JacksonDto, io.vertx.core.json.JsonObject> method23() {
            return null;
        }

        @Incoming("channel26")
        @Outgoing("channel27")
        ProcessorBuilder<Message<JacksonDto>, Message<io.vertx.core.json.JsonObject>> method24() {
            return null;
        }

        @Incoming("channel28")
        @Outgoing("channel29")
        ProcessorBuilder<JacksonDto, io.vertx.core.json.JsonObject> method25() {
            return null;
        }

        @Incoming("channel30")
        @Outgoing("channel31")
        Publisher<Message<io.vertx.core.json.JsonObject>> method26(Message<JacksonDto> msg) {
            return null;
        }

        @Incoming("channel32")
        @Outgoing("channel33")
        Publisher<io.vertx.core.json.JsonObject> method27(JacksonDto payload) {
            return null;
        }

        @Incoming("channel34")
        @Outgoing("channel35")
        PublisherBuilder<Message<io.vertx.core.json.JsonObject>> method28(Message<JacksonDto> msg) {
            return null;
        }

        @Incoming("channel36")
        @Outgoing("channel37")
        PublisherBuilder<io.vertx.core.json.JsonObject> method29(JacksonDto payload) {
            return null;
        }

        @Incoming("channel38")
        @Outgoing("channel39")
        Multi<Message<io.vertx.core.json.JsonObject>> method30(Message<JacksonDto> msg) {
            return null;
        }

        @Incoming("channel40")
        @Outgoing("channel41")
        Multi<io.vertx.core.json.JsonObject> method31(JacksonDto payload) {
            return null;
        }

        @Incoming("channel42")
        @Outgoing("channel43")
        Message<io.vertx.core.json.JsonObject> method32(Message<JacksonDto> msg) {
            return null;
        }

        @Incoming("channel44")
        @Outgoing("channel45")
        io.vertx.core.json.JsonObject method33(JacksonDto payload) {
            return null;
        }

        @Incoming("channel46")
        @Outgoing("channel47")
        CompletionStage<Message<io.vertx.core.json.JsonObject>> method34(Message<JacksonDto> msg) {
            return null;
        }

        @Incoming("channel48")
        @Outgoing("channel49")
        CompletionStage<io.vertx.core.json.JsonObject> method35(JacksonDto payload) {
            return null;
        }

        @Incoming("channel50")
        @Outgoing("channel51")
        Uni<Message<io.vertx.core.json.JsonObject>> method36(Message<JacksonDto> msg) {
            return null;
        }

        @Incoming("channel52")
        @Outgoing("channel53")
        Uni<io.vertx.core.json.JsonObject> method37(JacksonDto payload) {
            return null;
        }

        // @Incoming @Outgoing stream manipulation

        @Incoming("channel54")
        @Outgoing("channel55")
        Publisher<Message<io.vertx.core.json.JsonObject>> method38(Publisher<Message<JacksonDto>> msg) {
            return null;
        }

        @Incoming("channel56")
        @Outgoing("channel57")
        Publisher<io.vertx.core.json.JsonObject> method39(Publisher<JacksonDto> payload) {
            return null;
        }

        @Incoming("channel58")
        @Outgoing("channel59")
        PublisherBuilder<Message<io.vertx.core.json.JsonObject>> method40(PublisherBuilder<Message<JacksonDto>> msg) {
            return null;
        }

        @Incoming("channel60")
        @Outgoing("channel61")
        PublisherBuilder<io.vertx.core.json.JsonObject> method41(PublisherBuilder<JacksonDto> payload) {
            return null;
        }

        @Incoming("channel62")
        @Outgoing("channel63")
        Multi<Message<io.vertx.core.json.JsonObject>> method42(Multi<Message<JacksonDto>> msg) {
            return null;
        }

        @Incoming("channel64")
        @Outgoing("channel65")
        Multi<io.vertx.core.json.JsonObject> method43(Multi<JacksonDto> payload) {
            return null;
        }
    }

    // ---

    @Test
    public void kafkaBytesInJsonbDtoOut() {
        // @formatter:off
        Tuple[] expectations = {
                tuple("mp.messaging.outgoing.channel1.value.serializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.outgoing.channel2.value.serializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.outgoing.channel3.value.serializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.outgoing.channel4.value.serializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.outgoing.channel5.value.serializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.outgoing.channel6.value.serializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.outgoing.channel7.value.serializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.outgoing.channel8.value.serializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.outgoing.channel9.value.serializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.outgoing.channel10.value.serializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.outgoing.channel11.value.serializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.outgoing.channel12.value.serializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),

                tuple("mp.messaging.incoming.channel13.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.incoming.channel14.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.incoming.channel15.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.incoming.channel16.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.incoming.channel17.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.incoming.channel18.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.incoming.channel19.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.incoming.channel20.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.incoming.channel21.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),

                tuple("mp.messaging.incoming.channel22.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel23.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel24.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel25.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel26.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel27.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel28.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel29.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel30.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel31.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel32.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel33.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel34.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel35.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel36.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel37.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel38.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel39.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel40.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel41.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel42.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel43.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel44.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel45.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel46.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel47.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel48.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel49.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel50.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel51.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel52.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel53.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),

                tuple("mp.messaging.incoming.channel54.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel55.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel56.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel57.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel58.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel59.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel60.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel61.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel62.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel63.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel64.value.deserializer", "org.apache.kafka.common.serialization.BytesDeserializer"),
                tuple("mp.messaging.outgoing.channel65.value.serializer",   "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
        };
        // @formatter:on

        doTest(expectations, JsonbDto.class, JsonbDtoSerializer.class, KafkaBytesInJsonbDtoOut.class);
    }

    static class JsonbDto {
    }

    static class JsonbDtoSerializer extends JsonbSerializer<JsonbDto> {
        @Override
        public byte[] serialize(String topic, Headers headers, JsonbDto data) {
            return null;
        }
    }

    private static class KafkaBytesInJsonbDtoOut {
        // @Outgoing

        @Outgoing("channel1")
        Publisher<Message<JsonbDto>> method1() {
            return null;
        }

        @Outgoing("channel2")
        Publisher<JsonbDto> method2() {
            return null;
        }

        @Outgoing("channel3")
        PublisherBuilder<Message<JsonbDto>> method3() {
            return null;
        }

        @Outgoing("channel4")
        PublisherBuilder<JsonbDto> method4() {
            return null;
        }

        @Outgoing("channel5")
        Multi<Message<JsonbDto>> method5() {
            return null;
        }

        @Outgoing("channel6")
        Multi<JsonbDto> method6() {
            return null;
        }

        @Outgoing("channel7")
        Message<JsonbDto> method7() {
            return null;
        }

        @Outgoing("channel8")
        JsonbDto method8() {
            return null;
        }

        @Outgoing("channel9")
        CompletionStage<Message<JsonbDto>> method9() {
            return null;
        }

        @Outgoing("channel10")
        CompletionStage<JsonbDto> method10() {
            return null;
        }

        @Outgoing("channel11")
        Uni<Message<JsonbDto>> method11() {
            return null;
        }

        @Outgoing("channel12")
        Uni<JsonbDto> method12() {
            return null;
        }

        // @Incoming

        @Incoming("channel13")
        Subscriber<Message<org.apache.kafka.common.utils.Bytes>> method13() {
            return null;
        }

        @Incoming("channel14")
        Subscriber<org.apache.kafka.common.utils.Bytes> method14() {
            return null;
        }

        @Incoming("channel15")
        SubscriberBuilder<Message<org.apache.kafka.common.utils.Bytes>, Void> method15() {
            return null;
        }

        @Incoming("channel16")
        SubscriberBuilder<org.apache.kafka.common.utils.Bytes, Void> method16() {
            return null;
        }

        @Incoming("channel17")
        void method17(org.apache.kafka.common.utils.Bytes msg) {
        }

        @Incoming("channel18")
        CompletionStage<?> method18(Message<org.apache.kafka.common.utils.Bytes> msg) {
            return null;
        }

        @Incoming("channel19")
        CompletionStage<?> method19(org.apache.kafka.common.utils.Bytes payload) {
            return null;
        }

        @Incoming("channel20")
        Uni<?> method20(Message<org.apache.kafka.common.utils.Bytes> msg) {
            return null;
        }

        @Incoming("channel21")
        Uni<?> method21(org.apache.kafka.common.utils.Bytes payload) {
            return null;
        }

        // @Incoming @Outgoing

        @Incoming("channel22")
        @Outgoing("channel23")
        Processor<Message<org.apache.kafka.common.utils.Bytes>, Message<JsonbDto>> method22() {
            return null;
        }

        @Incoming("channel24")
        @Outgoing("channel25")
        Processor<org.apache.kafka.common.utils.Bytes, JsonbDto> method23() {
            return null;
        }

        @Incoming("channel26")
        @Outgoing("channel27")
        ProcessorBuilder<Message<org.apache.kafka.common.utils.Bytes>, Message<JsonbDto>> method24() {
            return null;
        }

        @Incoming("channel28")
        @Outgoing("channel29")
        ProcessorBuilder<org.apache.kafka.common.utils.Bytes, JsonbDto> method25() {
            return null;
        }

        @Incoming("channel30")
        @Outgoing("channel31")
        Publisher<Message<JsonbDto>> method26(Message<org.apache.kafka.common.utils.Bytes> msg) {
            return null;
        }

        @Incoming("channel32")
        @Outgoing("channel33")
        Publisher<JsonbDto> method27(org.apache.kafka.common.utils.Bytes payload) {
            return null;
        }

        @Incoming("channel34")
        @Outgoing("channel35")
        PublisherBuilder<Message<JsonbDto>> method28(Message<org.apache.kafka.common.utils.Bytes> msg) {
            return null;
        }

        @Incoming("channel36")
        @Outgoing("channel37")
        PublisherBuilder<JsonbDto> method29(org.apache.kafka.common.utils.Bytes payload) {
            return null;
        }

        @Incoming("channel38")
        @Outgoing("channel39")
        Multi<Message<JsonbDto>> method30(Message<org.apache.kafka.common.utils.Bytes> msg) {
            return null;
        }

        @Incoming("channel40")
        @Outgoing("channel41")
        Multi<JsonbDto> method31(org.apache.kafka.common.utils.Bytes payload) {
            return null;
        }

        @Incoming("channel42")
        @Outgoing("channel43")
        Message<JsonbDto> method32(Message<org.apache.kafka.common.utils.Bytes> msg) {
            return null;
        }

        @Incoming("channel44")
        @Outgoing("channel45")
        JsonbDto method33(org.apache.kafka.common.utils.Bytes payload) {
            return null;
        }

        @Incoming("channel46")
        @Outgoing("channel47")
        CompletionStage<Message<JsonbDto>> method34(Message<org.apache.kafka.common.utils.Bytes> msg) {
            return null;
        }

        @Incoming("channel48")
        @Outgoing("channel49")
        CompletionStage<JsonbDto> method35(org.apache.kafka.common.utils.Bytes payload) {
            return null;
        }

        @Incoming("channel50")
        @Outgoing("channel51")
        Uni<Message<JsonbDto>> method36(Message<org.apache.kafka.common.utils.Bytes> msg) {
            return null;
        }

        @Incoming("channel52")
        @Outgoing("channel53")
        Uni<JsonbDto> method37(org.apache.kafka.common.utils.Bytes payload) {
            return null;
        }

        // @Incoming @Outgoing stream manipulation

        @Incoming("channel54")
        @Outgoing("channel55")
        Publisher<Message<JsonbDto>> method38(Publisher<Message<org.apache.kafka.common.utils.Bytes>> msg) {
            return null;
        }

        @Incoming("channel56")
        @Outgoing("channel57")
        Publisher<JsonbDto> method39(Publisher<org.apache.kafka.common.utils.Bytes> payload) {
            return null;
        }

        @Incoming("channel58")
        @Outgoing("channel59")
        PublisherBuilder<Message<JsonbDto>> method40(PublisherBuilder<Message<org.apache.kafka.common.utils.Bytes>> msg) {
            return null;
        }

        @Incoming("channel60")
        @Outgoing("channel61")
        PublisherBuilder<JsonbDto> method41(PublisherBuilder<org.apache.kafka.common.utils.Bytes> payload) {
            return null;
        }

        @Incoming("channel62")
        @Outgoing("channel63")
        Multi<Message<JsonbDto>> method42(Multi<Message<org.apache.kafka.common.utils.Bytes>> msg) {
            return null;
        }

        @Incoming("channel64")
        @Outgoing("channel65")
        Multi<JsonbDto> method43(Multi<org.apache.kafka.common.utils.Bytes> payload) {
            return null;
        }
    }

    // ---

    @Test
    public void kafkaRecordIntUuidInRecordDoubleByteBufferOut() {
        // @formatter:off
        Tuple[] expectations = {
                tuple("mp.messaging.outgoing.channel1.key.serializer",    "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel1.value.serializer",  "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.outgoing.channel3.key.serializer",    "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel3.value.serializer",  "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.outgoing.channel5.key.serializer",    "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel5.value.serializer",  "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.outgoing.channel7.key.serializer",    "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel7.value.serializer",  "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.outgoing.channel9.key.serializer",    "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel9.value.serializer",  "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.outgoing.channel11.key.serializer",   "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel11.value.serializer", "org.apache.kafka.common.serialization.ByteBufferSerializer"),

                tuple("mp.messaging.incoming.channel13.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel13.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.incoming.channel15.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel15.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.incoming.channel18.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel18.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.incoming.channel20.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel20.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),

                tuple("mp.messaging.incoming.channel22.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel22.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel23.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel23.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.incoming.channel26.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel26.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel27.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel27.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.incoming.channel30.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel30.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel31.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel31.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.incoming.channel34.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel34.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel35.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel35.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.incoming.channel38.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel38.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel39.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel39.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.incoming.channel42.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel42.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel43.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel43.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.incoming.channel46.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel46.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel47.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel47.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.incoming.channel50.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel50.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel51.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel51.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),

                tuple("mp.messaging.incoming.channel54.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel54.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel55.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel55.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.incoming.channel58.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel58.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel59.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel59.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.incoming.channel62.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel62.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel63.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel63.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
        };
        // @formatter:on

        doTest(expectations, KafkaRecordIntUuidInRecordDoubleByteBufferOut.class);
    }

    private static class KafkaRecordIntUuidInRecordDoubleByteBufferOut {
        // @Outgoing

        @Outgoing("channel1")
        Publisher<Record<Double, java.nio.ByteBuffer>> method1() {
            return null;
        }

        @Outgoing("channel3")
        PublisherBuilder<Record<Double, java.nio.ByteBuffer>> method3() {
            return null;
        }

        @Outgoing("channel5")
        Multi<Record<Double, java.nio.ByteBuffer>> method5() {
            return null;
        }

        @Outgoing("channel7")
        Record<Double, java.nio.ByteBuffer> method7() {
            return null;
        }

        @Outgoing("channel9")
        CompletionStage<Record<Double, java.nio.ByteBuffer>> method9() {
            return null;
        }

        @Outgoing("channel11")
        Uni<Record<Double, java.nio.ByteBuffer>> method11() {
            return null;
        }

        // @Incoming

        @Incoming("channel13")
        Subscriber<KafkaRecord<Integer, java.util.UUID>> method13() {
            return null;
        }

        @Incoming("channel15")
        SubscriberBuilder<KafkaRecord<Integer, java.util.UUID>, Void> method15() {
            return null;
        }

        @Incoming("channel18")
        CompletionStage<?> method18(KafkaRecord<Integer, java.util.UUID> msg) {
            return null;
        }

        @Incoming("channel20")
        Uni<?> method20(KafkaRecord<Integer, java.util.UUID> msg) {
            return null;
        }

        // @Incoming @Outgoing

        @Incoming("channel22")
        @Outgoing("channel23")
        Processor<KafkaRecord<Integer, java.util.UUID>, Record<Double, java.nio.ByteBuffer>> method22() {
            return null;
        }

        @Incoming("channel26")
        @Outgoing("channel27")
        ProcessorBuilder<KafkaRecord<Integer, java.util.UUID>, Record<Double, java.nio.ByteBuffer>> method24() {
            return null;
        }

        @Incoming("channel30")
        @Outgoing("channel31")
        Publisher<Record<Double, java.nio.ByteBuffer>> method26(KafkaRecord<Integer, java.util.UUID> msg) {
            return null;
        }

        @Incoming("channel34")
        @Outgoing("channel35")
        PublisherBuilder<Record<Double, java.nio.ByteBuffer>> method28(KafkaRecord<Integer, java.util.UUID> msg) {
            return null;
        }

        @Incoming("channel38")
        @Outgoing("channel39")
        Multi<Record<Double, java.nio.ByteBuffer>> method30(KafkaRecord<Integer, java.util.UUID> msg) {
            return null;
        }

        @Incoming("channel42")
        @Outgoing("channel43")
        Record<Double, java.nio.ByteBuffer> method32(KafkaRecord<Integer, java.util.UUID> msg) {
            return null;
        }

        @Incoming("channel46")
        @Outgoing("channel47")
        CompletionStage<Record<Double, java.nio.ByteBuffer>> method34(KafkaRecord<Integer, java.util.UUID> msg) {
            return null;
        }

        @Incoming("channel50")
        @Outgoing("channel51")
        Uni<Record<Double, java.nio.ByteBuffer>> method36(KafkaRecord<Integer, java.util.UUID> msg) {
            return null;
        }

        // @Incoming @Outgoing stream manipulation

        @Incoming("channel54")
        @Outgoing("channel55")
        Publisher<Record<Double, java.nio.ByteBuffer>> method38(Publisher<KafkaRecord<Integer, java.util.UUID>> msg) {
            return null;
        }

        @Incoming("channel58")
        @Outgoing("channel59")
        PublisherBuilder<Record<Double, java.nio.ByteBuffer>> method40(
                PublisherBuilder<KafkaRecord<Integer, java.util.UUID>> msg) {
            return null;
        }

        @Incoming("channel62")
        @Outgoing("channel63")
        Multi<Record<Double, java.nio.ByteBuffer>> method42(Multi<KafkaRecord<Integer, java.util.UUID>> msg) {
            return null;
        }
    }

    // ---

    @Test
    public void consumerRecordIntUuidInProducerRecordDoubleByteBufferOut() {
        // @formatter:off
        Tuple[] expectations = {
                tuple("mp.messaging.outgoing.channel1.key.serializer",    "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel1.value.serializer",  "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.outgoing.channel3.key.serializer",    "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel3.value.serializer",  "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.outgoing.channel5.key.serializer",    "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel5.value.serializer",  "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.outgoing.channel7.key.serializer",    "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel7.value.serializer",  "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.outgoing.channel9.key.serializer",    "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel9.value.serializer",  "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.outgoing.channel11.key.serializer",   "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel11.value.serializer", "org.apache.kafka.common.serialization.ByteBufferSerializer"),

                tuple("mp.messaging.incoming.channel13.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel13.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.incoming.channel15.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel15.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.incoming.channel18.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel18.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.incoming.channel20.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel20.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),

                tuple("mp.messaging.incoming.channel22.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel22.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel23.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel23.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.incoming.channel26.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel26.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel27.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel27.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.incoming.channel30.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel30.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel31.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel31.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.incoming.channel34.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel34.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel35.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel35.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.incoming.channel38.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel38.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel39.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel39.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.incoming.channel42.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel42.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel43.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel43.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.incoming.channel46.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel46.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel47.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel47.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.incoming.channel50.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel50.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel51.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel51.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),

                tuple("mp.messaging.incoming.channel54.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel54.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel55.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel55.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.incoming.channel58.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel58.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel59.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel59.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
                tuple("mp.messaging.incoming.channel62.key.deserializer",   "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel62.value.deserializer", "org.apache.kafka.common.serialization.UUIDDeserializer"),
                tuple("mp.messaging.outgoing.channel63.key.serializer",     "org.apache.kafka.common.serialization.DoubleSerializer"),
                tuple("mp.messaging.outgoing.channel63.value.serializer",   "org.apache.kafka.common.serialization.ByteBufferSerializer"),
        };
        // @formatter:on

        doTest(expectations, ConsumerRecordIntUuidInProducerRecordDoubleByteBufferOut.class);
    }

    private static class ConsumerRecordIntUuidInProducerRecordDoubleByteBufferOut {
        // @Outgoing

        @Outgoing("channel1")
        Publisher<ProducerRecord<Double, java.nio.ByteBuffer>> method1() {
            return null;
        }

        @Outgoing("channel3")
        PublisherBuilder<ProducerRecord<Double, java.nio.ByteBuffer>> method3() {
            return null;
        }

        @Outgoing("channel5")
        Multi<ProducerRecord<Double, java.nio.ByteBuffer>> method5() {
            return null;
        }

        @Outgoing("channel7")
        ProducerRecord<Double, java.nio.ByteBuffer> method7() {
            return null;
        }

        @Outgoing("channel9")
        CompletionStage<ProducerRecord<Double, java.nio.ByteBuffer>> method9() {
            return null;
        }

        @Outgoing("channel11")
        Uni<ProducerRecord<Double, java.nio.ByteBuffer>> method11() {
            return null;
        }

        // @Incoming

        @Incoming("channel13")
        Subscriber<ConsumerRecord<Integer, java.util.UUID>> method13() {
            return null;
        }

        @Incoming("channel15")
        SubscriberBuilder<ConsumerRecord<Integer, java.util.UUID>, Void> method15() {
            return null;
        }

        @Incoming("channel18")
        CompletionStage<?> method18(ConsumerRecord<Integer, java.util.UUID> msg) {
            return null;
        }

        @Incoming("channel20")
        Uni<?> method20(ConsumerRecord<Integer, java.util.UUID> msg) {
            return null;
        }

        // @Incoming @Outgoing

        @Incoming("channel22")
        @Outgoing("channel23")
        Processor<ConsumerRecord<Integer, java.util.UUID>, ProducerRecord<Double, java.nio.ByteBuffer>> method22() {
            return null;
        }

        @Incoming("channel26")
        @Outgoing("channel27")
        ProcessorBuilder<ConsumerRecord<Integer, java.util.UUID>, ProducerRecord<Double, java.nio.ByteBuffer>> method24() {
            return null;
        }

        @Incoming("channel30")
        @Outgoing("channel31")
        Publisher<ProducerRecord<Double, java.nio.ByteBuffer>> method26(ConsumerRecord<Integer, java.util.UUID> msg) {
            return null;
        }

        @Incoming("channel34")
        @Outgoing("channel35")
        PublisherBuilder<ProducerRecord<Double, java.nio.ByteBuffer>> method28(ConsumerRecord<Integer, java.util.UUID> msg) {
            return null;
        }

        @Incoming("channel38")
        @Outgoing("channel39")
        Multi<ProducerRecord<Double, java.nio.ByteBuffer>> method30(ConsumerRecord<Integer, java.util.UUID> msg) {
            return null;
        }

        @Incoming("channel42")
        @Outgoing("channel43")
        ProducerRecord<Double, java.nio.ByteBuffer> method32(ConsumerRecord<Integer, java.util.UUID> msg) {
            return null;
        }

        @Incoming("channel46")
        @Outgoing("channel47")
        CompletionStage<ProducerRecord<Double, java.nio.ByteBuffer>> method34(ConsumerRecord<Integer, java.util.UUID> msg) {
            return null;
        }

        @Incoming("channel50")
        @Outgoing("channel51")
        Uni<ProducerRecord<Double, java.nio.ByteBuffer>> method36(ConsumerRecord<Integer, java.util.UUID> msg) {
            return null;
        }

        // @Incoming @Outgoing stream manipulation

        @Incoming("channel54")
        @Outgoing("channel55")
        Publisher<ProducerRecord<Double, java.nio.ByteBuffer>> method38(
                Publisher<ConsumerRecord<Integer, java.util.UUID>> msg) {
            return null;
        }

        @Incoming("channel58")
        @Outgoing("channel59")
        PublisherBuilder<ProducerRecord<Double, java.nio.ByteBuffer>> method40(
                PublisherBuilder<ConsumerRecord<Integer, java.util.UUID>> msg) {
            return null;
        }

        @Incoming("channel62")
        @Outgoing("channel63")
        Multi<ProducerRecord<Double, java.nio.ByteBuffer>> method42(Multi<ConsumerRecord<Integer, java.util.UUID>> msg) {
            return null;
        }
    }

    // ---

    @Test
    public void floatJsonArrayInShortByteArrayOut() {
        // @formatter:off
        Tuple[] expectations = {
                tuple("mp.messaging.outgoing.channel1.value.serializer",  "org.apache.kafka.common.serialization.ByteArraySerializer"),
                tuple("mp.messaging.outgoing.channel2.value.serializer",  "org.apache.kafka.common.serialization.ByteArraySerializer"),
                tuple("mp.messaging.outgoing.channel3.key.serializer",    "org.apache.kafka.common.serialization.ShortSerializer"),
                tuple("mp.messaging.outgoing.channel3.value.serializer",  "org.apache.kafka.common.serialization.ByteArraySerializer"),
                tuple("mp.messaging.outgoing.channel4.key.serializer",    "org.apache.kafka.common.serialization.ShortSerializer"),
                tuple("mp.messaging.outgoing.channel4.value.serializer",  "org.apache.kafka.common.serialization.ByteArraySerializer"),
                tuple("mp.messaging.outgoing.channel5.key.serializer",    "org.apache.kafka.common.serialization.ShortSerializer"),
                tuple("mp.messaging.outgoing.channel5.value.serializer",  "org.apache.kafka.common.serialization.ByteArraySerializer"),
                tuple("mp.messaging.outgoing.channel6.value.serializer",  "org.apache.kafka.common.serialization.ByteArraySerializer"),
                tuple("mp.messaging.outgoing.channel7.value.serializer",  "org.apache.kafka.common.serialization.ByteArraySerializer"),
                tuple("mp.messaging.outgoing.channel8.key.serializer",    "org.apache.kafka.common.serialization.ShortSerializer"),
                tuple("mp.messaging.outgoing.channel8.value.serializer",  "org.apache.kafka.common.serialization.ByteArraySerializer"),
                tuple("mp.messaging.outgoing.channel9.key.serializer",    "org.apache.kafka.common.serialization.ShortSerializer"),
                tuple("mp.messaging.outgoing.channel9.value.serializer",  "org.apache.kafka.common.serialization.ByteArraySerializer"),
                tuple("mp.messaging.outgoing.channel10.key.serializer",    "org.apache.kafka.common.serialization.ShortSerializer"),
                tuple("mp.messaging.outgoing.channel10.value.serializer",  "org.apache.kafka.common.serialization.ByteArraySerializer"),

                tuple("mp.messaging.incoming.channel11.value.deserializer",  "io.quarkus.kafka.client.serialization.JsonArrayDeserializer"),
                tuple("mp.messaging.incoming.channel12.value.deserializer", "io.quarkus.kafka.client.serialization.JsonArrayDeserializer"),
                tuple("mp.messaging.incoming.channel13.key.deserializer",   "org.apache.kafka.common.serialization.FloatDeserializer"),
                tuple("mp.messaging.incoming.channel13.value.deserializer", "io.quarkus.kafka.client.serialization.JsonArrayDeserializer"),
                tuple("mp.messaging.incoming.channel14.key.deserializer",   "org.apache.kafka.common.serialization.FloatDeserializer"),
                tuple("mp.messaging.incoming.channel14.value.deserializer", "io.quarkus.kafka.client.serialization.JsonArrayDeserializer"),
                tuple("mp.messaging.incoming.channel15.key.deserializer",   "org.apache.kafka.common.serialization.FloatDeserializer"),
                tuple("mp.messaging.incoming.channel15.value.deserializer", "io.quarkus.kafka.client.serialization.JsonArrayDeserializer"),
                tuple("mp.messaging.incoming.channel16.value.deserializer", "io.quarkus.kafka.client.serialization.JsonArrayDeserializer"),
                tuple("mp.messaging.incoming.channel17.value.deserializer", "io.quarkus.kafka.client.serialization.JsonArrayDeserializer"),
                tuple("mp.messaging.incoming.channel18.key.deserializer",   "org.apache.kafka.common.serialization.FloatDeserializer"),
                tuple("mp.messaging.incoming.channel18.value.deserializer", "io.quarkus.kafka.client.serialization.JsonArrayDeserializer"),
                tuple("mp.messaging.incoming.channel19.key.deserializer",   "org.apache.kafka.common.serialization.FloatDeserializer"),
                tuple("mp.messaging.incoming.channel19.value.deserializer", "io.quarkus.kafka.client.serialization.JsonArrayDeserializer"),
                tuple("mp.messaging.incoming.channel20.key.deserializer",   "org.apache.kafka.common.serialization.FloatDeserializer"),
                tuple("mp.messaging.incoming.channel20.value.deserializer", "io.quarkus.kafka.client.serialization.JsonArrayDeserializer"),
                tuple("mp.messaging.incoming.channel21.value.deserializer", "io.quarkus.kafka.client.serialization.JsonArrayDeserializer"),
                tuple("mp.messaging.incoming.channel22.value.deserializer", "io.quarkus.kafka.client.serialization.JsonArrayDeserializer"),
                tuple("mp.messaging.incoming.channel23.key.deserializer",   "org.apache.kafka.common.serialization.FloatDeserializer"),
                tuple("mp.messaging.incoming.channel23.value.deserializer", "io.quarkus.kafka.client.serialization.JsonArrayDeserializer"),
                tuple("mp.messaging.incoming.channel24.key.deserializer",   "org.apache.kafka.common.serialization.FloatDeserializer"),
                tuple("mp.messaging.incoming.channel24.value.deserializer", "io.quarkus.kafka.client.serialization.JsonArrayDeserializer"),
                tuple("mp.messaging.incoming.channel25.key.deserializer",   "org.apache.kafka.common.serialization.FloatDeserializer"),
                tuple("mp.messaging.incoming.channel25.value.deserializer", "io.quarkus.kafka.client.serialization.JsonArrayDeserializer"),
        };
        // @formatter:on

        doTest(expectations, FloatJsonArrayInShortByteArrayOut.class);
    }

    private static class FloatJsonArrayInShortByteArrayOut {
        // outgoing

        @Inject
        @Channel("channel1")
        Emitter<byte[]> emitter1;

        @Inject
        @Channel("channel2")
        Emitter<Message<byte[]>> emitter2;

        @Inject
        @Channel("channel3")
        Emitter<KafkaRecord<Short, byte[]>> emitter3;

        @Inject
        @Channel("channel4")
        Emitter<Record<Short, byte[]>> emitter4;

        @Inject
        @Channel("channel5")
        Emitter<ProducerRecord<Short, byte[]>> emitter5;

        @Inject
        @Channel("channel6")
        MutinyEmitter<byte[]> emitter6;

        @Inject
        @Channel("channel7")
        MutinyEmitter<Message<byte[]>> emitter7;

        @Inject
        @Channel("channel8")
        MutinyEmitter<KafkaRecord<Short, byte[]>> emitter8;

        @Inject
        @Channel("channel9")
        MutinyEmitter<Record<Short, byte[]>> emitter9;

        @Inject
        @Channel("channel10")
        MutinyEmitter<ProducerRecord<Short, byte[]>> emitter10;

        // incoming

        @Inject
        @Channel("channel11")
        Publisher<JsonArray> consumer11;

        @Inject
        @Channel("channel12")
        Publisher<Message<JsonArray>> consumer12;

        @Inject
        @Channel("channel13")
        Publisher<KafkaRecord<Float, JsonArray>> consumer13;

        @Inject
        @Channel("channel14")
        Publisher<Record<Float, JsonArray>> consumer14;

        @Inject
        @Channel("channel15")
        Publisher<ConsumerRecord<Float, JsonArray>> consumer15;

        @Inject
        @Channel("channel16")
        PublisherBuilder<JsonArray> consumer16;

        @Inject
        @Channel("channel17")
        PublisherBuilder<Message<JsonArray>> consumer17;

        @Inject
        @Channel("channel18")
        PublisherBuilder<KafkaRecord<Float, JsonArray>> consumer18;

        @Inject
        @Channel("channel19")
        PublisherBuilder<Record<Float, JsonArray>> consumer19;

        @Inject
        @Channel("channel20")
        PublisherBuilder<ConsumerRecord<Float, JsonArray>> consumer20;

        @Inject
        @Channel("channel21")
        Multi<JsonArray> consumer21;

        @Inject
        @Channel("channel22")
        Multi<Message<JsonArray>> consumer22;

        @Inject
        @Channel("channel23")
        Multi<KafkaRecord<Float, JsonArray>> consumer23;

        @Inject
        @Channel("channel24")
        Multi<Record<Float, JsonArray>> consumer24;

        @Inject
        @Channel("channel25")
        Multi<ConsumerRecord<Float, JsonArray>> consumer25;
    }

    @Test
    void produceDefaultConfigOnce() {
        // @formatter:off
        Tuple[] expectations = {
                tuple("mp.messaging.outgoing.channel1.value.serializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
                tuple("mp.messaging.incoming.channel2.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.incoming.channel3.key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel3.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
                tuple("mp.messaging.outgoing.channel4.key.serializer", "org.apache.kafka.common.serialization.StringSerializer"),
                tuple("mp.messaging.outgoing.channel4.value.serializer", "org.apache.kafka.common.serialization.IntegerSerializer"),
        };
        // @formatter:on

        doTest(expectations, JsonbDto.class, JsonbDtoSerializer.class, JacksonDto.class, JacksonDtoDeserializer.class,
                MultipleChannels.class);
    }

    private static class MultipleChannels {

        @Channel("channel1")
        Emitter<JsonbDto> emitter1;

        @Outgoing("channel1")
        Publisher<Message<JsonbDto>> method1() {
            return null;
        }

        @Outgoing("channel1")
        Publisher<JsonbDto> method1Duplicate() {
            return null;
        }

        @Channel("channel2")
        Multi<JacksonDto> channel2;

        @Incoming("channel2")
        void channel2Duplicate(JacksonDto jacksonDto) {
        }

        @Channel("channel3")
        Multi<Record<Integer, JacksonDto>> channel3;

        @Incoming("channel3")
        void channel3Duplicate(Record<Integer, JacksonDto> jacksonDto) {
        }

        @Channel("channel4")
        Emitter<ProducerRecord<String, Integer>> emitterChannel4;

        @Outgoing("channel4")
        ProducerRecord<String, Integer> method4() {
            return null;
        }
    }

    @Test
    void produceBatchConfigWithSerdeAutodetect() {
        // @formatter:off
        Tuple[] expectations = {
                tuple("mp.messaging.incoming.channel1.value.deserializer", JacksonDtoDeserializer.class.getName()),
                tuple("mp.messaging.incoming.channel1.batch", "true"),
                tuple("mp.messaging.incoming.channel2.value.deserializer", JacksonDtoDeserializer.class.getName()),
                tuple("mp.messaging.incoming.channel2.batch", "true"),
                tuple("mp.messaging.incoming.channel3.key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel3.value.deserializer", JacksonDtoDeserializer.class.getName()),
                tuple("mp.messaging.incoming.channel3.batch", "true"),
                tuple("mp.messaging.incoming.channel4.key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.incoming.channel4.value.deserializer", JacksonDtoDeserializer.class.getName()),
                tuple("mp.messaging.incoming.channel4.batch", "true"),
                tuple("mp.messaging.incoming.channel5.key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.channel5.value.deserializer", JacksonDtoDeserializer.class.getName()),
                tuple("mp.messaging.incoming.channel5.batch", "true"),
                tuple("mp.messaging.incoming.channel6.key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.incoming.channel6.value.deserializer", JacksonDtoDeserializer.class.getName()),
                tuple("mp.messaging.incoming.channel6.batch", "true"),
                tuple("mp.messaging.incoming.channel7.value.deserializer", JacksonDtoDeserializer.class.getName()),
                tuple("mp.messaging.incoming.channel7.batch", "true"),
                tuple("mp.messaging.incoming.channel8.value.deserializer", JacksonDtoDeserializer.class.getName()),
                tuple("mp.messaging.incoming.channel8.batch", "true"),
                tuple("mp.messaging.incoming.channel9.batch", "true"),
                tuple("mp.messaging.incoming.channel10.batch", "true"),
        };
        // @formatter:on

        doTest(expectations, JacksonDto.class, JacksonDtoDeserializer.class, BatchChannels.class);
    }

    private static class BatchChannels {

        @Channel("channel1")
        Multi<List<JacksonDto>> channel1;

        @Incoming("channel2")
        void channel2(List<JacksonDto> jacksonDto) {
        }

        @Channel("channel3")
        Multi<KafkaRecordBatch<Integer, JacksonDto>> channel3;

        @Incoming("channel4")
        void channel4(KafkaRecordBatch<String, JacksonDto> jacksonDto) {
        }

        @Channel("channel5")
        Multi<ConsumerRecords<Integer, JacksonDto>> channel5;

        @Incoming("channel6")
        void channel6(ConsumerRecords<String, JacksonDto> jacksonDto) {
        }

        @Channel("channel7")
        Multi<Message<List<JacksonDto>>> channel7;

        @Incoming("channel8")
        void channel8(Message<List<JacksonDto>> jacksonDto) {
        }

        @Channel("channel9") // Not supported
        Multi<List<Message<JacksonDto>>> channel9;

        @Incoming("channel10") // Not supported
        void channel10(List<Message<JacksonDto>> jacksonDto) {
        }

    }

    // ---

    @Test
    public void connectorConfigNotOverridden() {
        // @formatter:off
        Tuple[] expectations1 = {
                // "mp.messaging.outgoing.channel1.key.serializer" NOT expected, connector config exists
                tuple("mp.messaging.outgoing.channel1.value.serializer",   "org.apache.kafka.common.serialization.StringSerializer"),

                tuple("mp.messaging.incoming.channel2.key.deserializer",   "org.apache.kafka.common.serialization.LongDeserializer"),
                // "mp.messaging.incoming.channel2.value.deserializer" NOT expected, connector config exists

                tuple("mp.messaging.incoming.channel3.key.deserializer",   "org.apache.kafka.common.serialization.LongDeserializer"),
                // "mp.messaging.incoming.channel3.value.deserializer" NOT expected, connector config exists
                // "mp.messaging.outgoing.channel4.key.serializer" NOT expected, connector config exists
                tuple("mp.messaging.outgoing.channel4.value.serializer",   "org.apache.kafka.common.serialization.StringSerializer"),
        };

        Tuple[] expectations2 = {
                tuple("mp.messaging.outgoing.channel1.key.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                // "mp.messaging.outgoing.channel1.value.serializer" NOT expected, connector config exists

                // "mp.messaging.incoming.channel2.key.deserializer" NOT expected, connector config exists
                tuple("mp.messaging.incoming.channel2.value.deserializer",   "org.apache.kafka.common.serialization.StringDeserializer"),

                // "mp.messaging.incoming.channel3.key.deserializer" NOT expected, connector config exists
                tuple("mp.messaging.incoming.channel3.value.deserializer",   "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.channel4.key.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
                // "mp.messaging.outgoing.channel4.value.serializer" NOT expected, connector config exists
        };
        // @formatter:on

        doTest(new SmallRyeConfigBuilder()
                .withSources(new MapBackedConfigSource("test", Map.of(
                        "mp.messaging.connector.smallrye-kafka.key.serializer", "foo.Bar",
                        "mp.messaging.connector.smallrye-kafka.value.deserializer", "foo.Baz")) {
                })
                .build(), expectations1, ConnectorConfigNotOverriden.class);

        doTest(new SmallRyeConfigBuilder()
                .withSources(new MapBackedConfigSource("test", Map.of(
                        "mp.messaging.connector.smallrye-kafka.key.deserializer", "foo.Bar",
                        "mp.messaging.connector.smallrye-kafka.value.serializer", "foo.Baz")) {
                })
                .build(), expectations2, ConnectorConfigNotOverriden.class);
    }

    private static class ConnectorConfigNotOverriden {
        @Outgoing("channel1")
        Record<Long, String> method1() {
            return null;
        }

        @Incoming("channel2")
        CompletionStage<?> method2(Record<Long, String> msg) {
            return null;
        }

        @Incoming("channel3")
        @Outgoing("channel4")
        Record<Long, String> method3(Record<Long, String> msg) {
            return null;
        }
    }

    // ---

    @Test
    public void genericSerdeImplementationAutoDetect() {
        // @formatter:off

        Tuple[] expectations1 = {
                tuple("mp.messaging.outgoing.channel1.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.incoming.channel2.key.deserializer", "org.apache.kafka.common.serialization.LongDeserializer"),
                tuple("mp.messaging.incoming.channel3.value.deserializer", assertStartsWith("io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$CustomDto_Deserializer_")),
                tuple("mp.messaging.incoming.channel2.value.deserializer", "io.quarkus.kafka.client.serialization.JsonObjectDeserializer"),
        };

        var generated1 = List.of(assertStartsWith("io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeConfigTest$CustomDto_Deserializer_"));
        var reflective1 = List.of(assertStartsWith("io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$CustomDto_Deserializer_"));

        Tuple[] expectations2 = {
                tuple("mp.messaging.outgoing.channel1.value.serializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$MySerializer"),

                tuple("mp.messaging.incoming.channel2.key.deserializer", "org.apache.kafka.common.serialization.LongDeserializer"),
                tuple("mp.messaging.incoming.channel2.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$MyDeserializer"),

                tuple("mp.messaging.incoming.channel3.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$CustomDtoDeserializer"),
        };

        Tuple[] expectations3 = {
                tuple("mp.messaging.outgoing.channel1.value.serializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$MySerializer"),

                tuple("mp.messaging.incoming.channel2.key.deserializer", "org.apache.kafka.common.serialization.LongDeserializer"),
                tuple("mp.messaging.incoming.channel2.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$MyDeserializer"),

                tuple("mp.messaging.incoming.channel3.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$MyObjectMapperDeserializer"),
        };
        Tuple[] expectations4 = {
                tuple("mp.messaging.outgoing.channel1.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"),

                tuple("mp.messaging.incoming.channel2.key.deserializer", "org.apache.kafka.common.serialization.LongDeserializer"),
                tuple("mp.messaging.incoming.channel2.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$MyDeserializer"),

                tuple("mp.messaging.incoming.channel3.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$MyObjectMapperDeserializer"),
        };
        // @formatter:on

        doTest(null, expectations1, generated1, reflective1, CustomSerdeImplementation.class, CustomDto.class);

        doTest(expectations2, CustomSerdeImplementation.class, CustomDto.class,
                MySerializer.class,
                MyDeserializer.class, MyObjectMapperDeserializer.class, CustomDtoDeserializer.class);

        doTest(expectations3, CustomSerdeImplementation.class, CustomDto.class,
                MySerializer.class,
                MyDeserializer.class, MyObjectMapperDeserializer.class);

        doTest(expectations4, CustomSerdeImplementation.class, CustomDto.class,
                MyWrongSerializer.class, CustomInterface.class,
                MyDeserializer.class, MyObjectMapperDeserializer.class);
    }

    private static class CustomDto {

    }

    private interface CustomInterface<T> {

    }

    private static class MyWrongSerializer implements Serializer<Integer>, CustomInterface<Long> {

        @Override
        public byte[] serialize(String topic, Integer data) {
            return new byte[0];
        }
    }

    private static class MySerializer implements Serializer<Long>, CustomInterface<Long> {

        @Override
        public byte[] serialize(String topic, Long data) {
            return new byte[0];
        }
    }

    private static class MyDeserializer implements Deserializer<JsonObject> {

        @Override
        public JsonObject deserialize(String topic, byte[] data) {
            return null;
        }
    }

    private static class MyObjectMapperDeserializer extends ObjectMapperDeserializer<CustomDto> {

        public MyObjectMapperDeserializer() {
            super(CustomDto.class);
        }
    }

    private static class CustomDtoDeserializer implements Deserializer<CustomDto> {

        @Override
        public CustomDto deserialize(String topic, byte[] data) {
            return null;
        }
    }

    private static class CustomSerdeImplementation {
        @Outgoing("channel1")
        Multi<Long> method1() {
            return null;
        }

        @Incoming("channel2")
        void method2(Record<Long, JsonObject> msg) {

        }

        @Incoming("channel3")
        void method3(CustomDto payload) {

        }
    }

    @Test
    void kafkaTransactions() {
        // @formatter:off
        Tuple[] expectations = {
                tuple("mp.messaging.outgoing.tx.value.serializer", "org.apache.kafka.common.serialization.StringSerializer"),
                tuple("mp.messaging.outgoing.tx.transactional.id", "${quarkus.application.name}-tx"),
                tuple("mp.messaging.outgoing.tx.enable.idempotence", "true"),
                tuple("mp.messaging.outgoing.tx.acks", "all"),
        };
        doTest(expectations, TransactionalProducer.class);

    }

    private static class TransactionalProducer {

        @Channel("tx")
        KafkaTransactions<String> kafkaTransactions;

    }

    @Test
    void repeatableIncomings() {
        Tuple[] expectations = {
                tuple("mp.messaging.incoming.channel1.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.incoming.channel2.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.incoming.channel3.value.deserializer", "io.quarkus.kafka.client.serialization.JsonObjectDeserializer"),
                tuple("mp.messaging.incoming.channel4.value.deserializer", "io.quarkus.kafka.client.serialization.JsonObjectDeserializer"),
        };
        doTest(expectations, RepeatableIncomingsChannels.class);
    }


    private static class RepeatableIncomingsChannels {

        @Incoming("channel1")
        @Incoming("channel2")
        void method1(String msg) {

        }

        @Incoming("channel3")
        @Incoming("channel4")
        void method2(JsonObject msg) {

        }

    }

    @Test
    void repeatableOutgoings() {
        Tuple[] expectations = {
                tuple("mp.messaging.outgoing.channel1.value.serializer", "org.apache.kafka.common.serialization.StringSerializer"),
                tuple("mp.messaging.outgoing.channel2.value.serializer", "org.apache.kafka.common.serialization.StringSerializer"),
                tuple("mp.messaging.outgoing.channel3.value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.outgoing.channel4.value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.outgoing.channel5.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.outgoing.channel6.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
        };
        doTest(expectations, RepeatableOutgoingsChannels.class);
    }

    private static class RepeatableOutgoingsChannels {

        @Outgoing("channel1")
        @Outgoing("channel2")
        String method1() {
            return null;
        }

        @Outgoing("channel3")
        @Outgoing("channel4")
        JsonObject method2() {
            return null;
        }

        enum T {

        }

        @Outgoing("channel5")
        @Outgoing("channel6")
        MultiSplitter<Long, T> method3() {
            return null;
        }

    }

    @Test
    void channelNameContainingDot() {
        Tuple[] expectations = {
                tuple("mp.messaging.incoming.\"new.channel\".key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.incoming.\"new.channel\".value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.\"new.channel.out\".key.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.outgoing.\"new.channel.out\".value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.outgoing.\"new.channel.out\".transactional.id", "${quarkus.application.name}-new.channel.out"),
                tuple("mp.messaging.outgoing.\"new.channel.out\".enable.idempotence", "true"),
                tuple("mp.messaging.outgoing.\"new.channel.out\".acks", "all"),
        };
        doTest(expectations, ChannelContainingDot.class);
    }


    private static class ChannelContainingDot {

        @Incoming("new.channel")
        void method1(KafkaRecord<Integer, String> msg) {

        }

        @Channel("new.channel.out")
        KafkaTransactions<ProducerRecord<Long, JsonObject>> transactions;

    }

    @Test
    void deadLetterQueue() {
        Tuple[] expectations = {
                tuple("mp.messaging.incoming.channel1.value.deserializer",
                        assertStartsWith("io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$CustomDto_Deserializer_")),
                tuple("mp.messaging.incoming.channel1.dead-letter-queue.value.serializer",
                        assertStartsWith("io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$CustomDto_Serializer_")),
                tuple("mp.messaging.incoming.channel2.key.deserializer",
                        assertStartsWith("io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$CustomDto_Deserializer_")),
                tuple("mp.messaging.incoming.channel2.value.deserializer",
                        assertStartsWith("io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$CustomDto_Deserializer_")),
                tuple("mp.messaging.incoming.channel2.dead-letter-queue.key.serializer",
                        assertStartsWith("io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$CustomDto_Serializer_")),
                tuple("mp.messaging.incoming.channel2.dead-letter-queue.value.serializer",
                        assertStartsWith("io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$CustomDto_Serializer_")),
        };
        var generated = List.of(
                assertStartsWith("io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeConfigTest$CustomDto_Deserializer_"),
                assertStartsWith("io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeConfigTest$CustomDto_Serializer_")
        );
        var reflective = List.of(
                assertStartsWith("io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$CustomDto_Deserializer_"),
                assertStartsWith("io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$CustomDto_Serializer_")
        );
        doTest(new SmallRyeConfigBuilder()
                .withSources(new MapBackedConfigSource("test", Map.of(
                        "mp.messaging.incoming.channel1.failure-strategy", "dead-letter-queue",
                        "mp.messaging.incoming.channel2.failure-strategy", "delayed-retry-topic")) {
                })
                .build(), expectations, generated, reflective, DeadLetterQueue.class);
    }

    private static class DeadLetterQueue {

        @Incoming("channel1")
        void method1(CustomDto msg) {

        }

        @Incoming("channel2")
        void method2(Record<CustomDto, CustomDto> msg) {

        }

    }


    @Test
    void targetedOutgoings() {
        Tuple[] expectations = {
                tuple("mp.messaging.incoming.channel1.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.incoming.channel2.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
        };
        doTest(expectations, TargetedOutgoings.class);
    }


    private static class TargetedOutgoings {

        @Incoming("channel1")
        @Outgoing("out1")
        @Outgoing("out2")
        Targeted method1(String msg) {
            return null;
        }

        @Incoming("channel2")
        @Outgoing("out3")
        @Outgoing("out4")
        TargetedMessages method2(String msg) {
            return null;
        }

    }

    @Test
    void kafkaRequestReply() {
        Tuple[] expectations = {
                tuple("mp.messaging.outgoing.channel1.value.serializer", "org.apache.kafka.common.serialization.StringSerializer"),
                tuple("mp.messaging.outgoing.channel1.reply.value.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.outgoing.channel2.key.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.outgoing.channel2.value.serializer", "org.apache.kafka.common.serialization.StringSerializer"),
                tuple("mp.messaging.outgoing.channel2.reply.value.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.outgoing.channel3.key.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.outgoing.channel3.value.serializer", "org.apache.kafka.common.serialization.StringSerializer"),
                tuple("mp.messaging.outgoing.channel3.reply.key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.outgoing.channel3.reply.value.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.outgoing.channel4.value.serializer", "io.apicurio.registry.serde.avro.AvroKafkaSerializer"),
                tuple("mp.messaging.outgoing.channel4.reply.key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.outgoing.channel4.reply.value.deserializer", "io.quarkus.kafka.client.serialization.JsonObjectDeserializer"),
        };
        doTest(expectations, AvroDto.class, RequestReply.class);
    }

    private static class RequestReply {

        @Inject
        @Channel("channel1")
        KafkaRequestReply<String, Integer> requestReply;

        @Inject
        @Channel("channel2")
        KafkaRequestReply<ProducerRecord<Long, String>, Integer> requestReply2;

        @Inject
        @Channel("channel3")
        KafkaRequestReply<Record<Long, String>, Record<Integer, Integer>> requestReply3;

        @Inject
        @Channel("channel4")
        KafkaRequestReply<AvroDto, ConsumerRecord<Integer, JsonObject>> requestReply4;

    }

    @Test
    void kafkaGenericPayload() {
        Tuple[] expectations = {
                tuple("mp.messaging.incoming.channel1.value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
                tuple("mp.messaging.outgoing.out1.key.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
                tuple("mp.messaging.outgoing.out1.value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
                tuple("mp.messaging.incoming.channel2.key.deserializer", "org.apache.kafka.common.serialization.LongDeserializer"),
                tuple("mp.messaging.incoming.channel2.value.deserializer", "io.quarkus.kafka.client.serialization.JsonObjectDeserializer"),
                tuple("mp.messaging.outgoing.channel3.value.serializer", "org.apache.kafka.common.serialization.IntegerSerializer"),
                tuple("mp.messaging.outgoing.channel4.key.serializer", "org.apache.kafka.common.serialization.StringSerializer"),
                tuple("mp.messaging.outgoing.channel4.value.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
        };
        doTest(expectations, GenericPayloadProducer.class);
    }

    private static class GenericPayloadProducer {
        @Incoming("channel1")
        @Outgoing("out1")
        GenericPayload<ProducerRecord<Long, JsonObject>> method1(String msg) {
            return null;
        }

        @Incoming("channel2")
        void method2(GenericPayload<Record<Long, JsonObject>> msg) {
        }

        @Outgoing("channel3")
        GenericPayload<Integer> method3() {
            return null;
        }


        @Outgoing("channel4")
        Multi<GenericPayload<ProducerRecord<String, Long>>> method4() {
            return null;
        }
    }

    @Test
    void inheritingSerdeClass() {
        Tuple[] expectations = {
                tuple("mp.messaging.outgoing.channel1.value.serializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$MyChildSerializer"),
                tuple("mp.messaging.incoming.channel2.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$MyChildDeserializer"),
        };
        doTest(expectations, ChannelChildSerializer.class, MyChildSerializer.class, ParentSerializer.class, MyChildDeserializer.class, ParentDeserializer.class);
    }

    private static class MyChildSerializer extends ParentSerializer<JsonbDto> {

    }

    private static abstract class ParentSerializer<T> implements Serializer<T> {

        @Override
        public byte[] serialize(String topic, T data) {
            return new byte[0];
        }
    }

    private static class MyChildDeserializer extends ParentDeserializer {

    }

    private static abstract class ParentDeserializer implements Deserializer<JsonbDto> {

        @Override
        public JsonbDto deserialize(String topic, byte[] data) {
            return null;
        }
    }

    private static class ChannelChildSerializer {
        @Channel("channel1")
        Emitter<JsonbDto> emitter1;

        @Channel("channel2")
        Multi<JsonbDto> channel2;
    }

    @Test
    void instanceInjectionPoint() {
        Tuple[] expectations = {
                tuple("mp.messaging.outgoing.channel1.value.serializer", "org.apache.kafka.common.serialization.StringSerializer"),
                tuple("mp.messaging.incoming.channel2.value.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer"),
                tuple("mp.messaging.outgoing.channel3.value.serializer", "org.apache.kafka.common.serialization.DoubleSerializer"),
        };
        doTest(expectations, InstanceInjectionPoint.class);
    }

    private static class InstanceInjectionPoint {
        @Channel("channel1")
        Instance<Emitter<String>> emitter1;

        @Channel("channel2")
        Instance<Multi<Integer>> channel2;

        @Channel("channel3")
        InjectableInstance<MutinyEmitter<Double>> channel3;
    }


}
